package cn.iotab.iotkit.transport.api.access.handler.adaptor;

import cn.iotab.iotkit.transport.api.access.device.HandlerManager;
import cn.iotab.iotkit.transport.api.access.handler.DeviceMessageSender;
import cn.iotab.iotkit.transport.api.access.handler.DeviceReplyListener;
import cn.iotab.iotkit.transport.api.access.handler.DeviceSessionHandler;
import cn.iotab.iotkit.transport.api.access.handler.impl.DefaultProductMessageHandler;
import cn.iotab.iotkit.transport.api.access.model.custom.ProtocolConstant;
import cn.iotab.iotkit.transport.api.access.model.exception.IncompleteMessageException;
import cn.iotab.iotkit.transport.api.access.model.message.CommonDeviceMessage;
import cn.iotab.iotkit.transport.api.access.model.message.CommonDeviceMessageReply;
import cn.iotab.iotkit.transport.api.access.model.message.DeviceMessage;
import cn.iotab.iotkit.transport.api.access.model.message.DeviceMessageReply;
import cn.iotab.iotkit.transport.api.access.model.message.OriginMessageReply;
import cn.iotab.iotkit.transport.api.common.message.msg.CommonPayload;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionEvent;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionInfo;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionReportInfo;
import cn.iotab.iotkit.transport.api.common.util.AsyncCallbackTemplate;
import com.alibaba.fastjson.JSON;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ToServerMsgHandlerAdaptor {

    private final HandlerManager handlerManager;
    private final DeviceMessageSender deviceMessageSender;
    private final DeviceReplyListener deviceReplyListener;
    //sessionManager对session的监听
    private final DeviceSessionHandler sessionStorageListener;
    //用户自定义对Session的监听
    private final DeviceSessionHandler userDeviceSessionHandler;

    public ToServerMsgHandlerAdaptor(HandlerManager handlerManager,
                                     DeviceMessageSender deviceMessageSender,
                                     DeviceReplyListener deviceReplyListener,
                                     DeviceSessionHandler sessionStorageListener) {
        this.handlerManager = handlerManager;
        this.deviceMessageSender = deviceMessageSender;
        this.deviceReplyListener = deviceReplyListener;
        this.sessionStorageListener = sessionStorageListener;
        this.userDeviceSessionHandler = handlerManager.getDeviceSessionHandler();
    }

    private DefaultProductMessageHandler getProductProcessor(String productKey) {
        DefaultProductMessageHandler processor = handlerManager.getHandlerProductHandlerOrDefault(productKey);
        if (processor == null) {
            throw new UnsupportedOperationException("unsupported product: " + productKey);
        }
        return processor;
    }

    public void onDeviceSessionStateChange(SessionInfo sessionInfo, SessionEvent msg) {
        sessionStorageListener.onDeviceSessionStateChange(sessionInfo, msg);
        //
        userDeviceSessionHandler.onDeviceSessionStateChange(sessionInfo, msg);
    }

    public void onSessionReport(SessionInfo sessionInfo, SessionReportInfo msg) {
        sessionStorageListener.onDeviceSessionReport(sessionInfo, msg);
        //
        userDeviceSessionHandler.onDeviceSessionReport(sessionInfo, msg);
    }


    public void onFromDeviceRequest(SessionInfo sessionInfo, CommonPayload request) {
        DefaultProductMessageHandler processor = getProductProcessor(request.getProductKey());
        CommonDeviceMessage msg;
        try {
            msg = DeviceMessageConverter.convertMessage(request.getProductKey(), request.getDeviceId(), request.getPayload());
        } catch (IncompleteMessageException e) {
            log.error("receive IncompleteMessage from {}_{}, err: {}, message: {}",
                    request.getProductKey(), request.getDeviceId(), e.getMessage(), request.getPayload());
            return;
        }
        log.debug("receive msg: {}", msg);
        ListenableFuture<DeviceMessageReply> replyFuture = processor.process(msg);
        //
        String sessionId = sessionInfo.getSessionId();
        if (msg.needAck()) {
            if (replyFuture != null) {
                AsyncCallbackTemplate.withCallback(replyFuture,
                        reply -> deviceMessageSender.sendReply(sessionId, reply),
                        ex -> onProcessDeviceRequestThrowable(sessionId, msg, ex),
                        MoreExecutors.directExecutor());
            } else {
                log.warn("message {} need ack, but server has no ack", msg);
                DeviceMessageReply failReply = CommonDeviceMessageReply.commonErrorReply(msg, ProtocolConstant.CODE_SERVER_ERROR, "server has no reply");
                deviceMessageSender.sendReply(sessionId, failReply);
            }
        }
    }

    private void onProcessDeviceRequestThrowable(String sessionId, DeviceMessage message, Throwable ex) {
        CommonDeviceMessageReply reply = new CommonDeviceMessageReply();
        reply.from(message);
        reply.setCode(ProtocolConstant.CODE_SERVER_ERROR);
        reply.setTimestamp(System.currentTimeMillis());
        String errMsg = ex.getMessage();
        reply.setErrorMsg(errMsg == null ? "unknown error" : errMsg);
        log.warn("process message {} throws an exception, ex={}, msg={}", message.getPrintKey(), ex.getClass().getSimpleName(), errMsg);
        deviceMessageSender.sendReply(sessionId, reply);
    }

    public void onFromDeviceResponse(CommonPayload response) {
        OriginMessageReply messageReply = OriginMessageReply.fromJson(JSON.parseObject(response.getPayload()));
        try {
            messageReply.validate();
        } catch (IncompleteMessageException e) {
            log.error("receive IncompleteMessageReply from {}_{}, err: {}, message:{}",
                    response.getProductKey(), response.getDeviceId(), e.getMessage(), response.getPayload());
            return;
        }
        CommonDeviceMessageReply reply = new CommonDeviceMessageReply();
        reply.fromOriginMessageReply(response.getProductKey(), response.getDeviceId(), messageReply);
        log.debug("receive msg: {}", reply);
        deviceReplyListener.onReply(reply);
    }
}
